package DianShang_2024.ds_02.tzgc


import org.apache.spark.ml.{Pipeline, PipelineModel}
import org.apache.spark.ml.feature.{OneHotEncoder, StandardScaler, VectorAssembler}
import org.apache.spark.ml.linalg.SparseVector
import org.apache.spark.ml.linalg.DenseVector
import org.apache.spark.sql.{Row, SparkSession}
import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.functions.{dense_rank, expr}

import java.util.Properties

object tzgc02 {
  def main(args: Array[String]): Unit = {
      /*
           2、根据Hive的dwd库中相关表或MySQL中shtd_store中相关商品表（sku_info），获取id、spu_id、price、weight、tm_id、category3_id
           这六个字段并进行数据预处理，对price、weight进行规范化(StandardScaler)处理，对spu_id、tm_id、category3_id进行one-hot编码处
           理（若该商品属于该品牌则置为1，否则置为0）,并按照id进行升序排序，在集群中输出第一条数据前10列（无需展示字段名），将结果截图粘贴至客
           户端桌面【Release\任务C提交结果.docx】中对应的任务序号下。
      字段	类型	中文含义	备注
      id	double	主键
      price	double	价格
      weight	double	重量
      spu_id#1	double	spu_id 1	若属于该spu_id，则内容为1否则为0
      spu_id#2	double	spu_id 2	若属于该spu_id，则内容为1否则为0
      .....	double
      tm_id#1	double	品牌1	若属于该品牌，则内容为1否则为0
      tm_id#2	double	品牌2	若属于该品牌，则内容为1否则为0
      ……	double
      category3_id#1	double	分类级别3 1	若属于该分类级别3，则内容为1否则为0
      category3_id#2	double	分类级别3 2	若属于该分类级别3，则内容为1否则为0
      ……

      结果格式如下：
      --------------------第一条数据前10列结果展示为：---------------------
      1.0,0.892346,1.72568,0.0,0.0,0.0,0.0,1.0,0.0,0.0
       */

    val spark=SparkSession.builder()
      .master("local[*]")
      .appName("第二套卷子特征工程第二题")
      .config("hive.exec.dynamic.partition.mode","nonstrict")
//      .config("spark.serializer","org.apache.spark.serializer.KryoSerializer")
//      .config("spark.sql.extensions","org.apache.spark.sql.hudi.HoodieSparkSessionExtension")
      .enableHiveSupport()
      .getOrCreate()

    val connect=new Properties()
    connect.setProperty("user","root")
    connect.setProperty("password","123456")
    connect.setProperty("driver","com.mysql.jdbc.Driver")

    //  1.之所以要对下面三个字段进行dense_rank()密集排序,是为了给每一行都分配一个唯一的索引，并且要从0开始，所以要-1，这是为了给one-hot编码
    //  处理做准备，因为one-hot编码处理的数据希望是连续的整数，这样可以有效工作
    //  2.其实这里的dense_rank排序的效果和使用StringIndexer的效果是一样的
    val sku_info=spark.read
      .jdbc("jdbc:mysql://192.168.40.110/shtd_store?useSSL=false","sku_info",connect)
      .select("id","spu_id","price","weight","tm_id","category3_id")
      .withColumn("spu_id_index",dense_rank().over(Window.orderBy("spu_id"))-1)
      .withColumn("tm_id_index",dense_rank().over(Window.orderBy("tm_id"))-1)
      .withColumn("category3_id_index",dense_rank().over(Window.orderBy("category3_id"))-1)

    //  上面准备工作做完了，就可以进行one-hot编码
    //  setDropLast(false):one-hot编码处理一般会去掉最后一行的结果，但是我们这里不需要删除
    val one_hot_encoder=new OneHotEncoder()
      .setInputCols(Array("spu_id_index","tm_id_index","category3_id_index"))
      .setOutputCols(Array("spu_id_hot","tm_id_hot","category3_id_hot"))
      .setDropLast(false)


    //  进行规范化之前需要对其转化为特征向量
    //  vectorAssemble:向量汇编程序
    val vectorAssembler01=new VectorAssembler()
      .setInputCols(Array("price"))
      .setOutputCol("price_v")

    val vectorAssembler02=new VectorAssembler()
      .setInputCols(Array("weight"))
      .setOutputCol("weight_v")

    //  标准化器，将特征向量标准化(和vectorAssembler一样，一个容器处理一个字段)
    //  setWithMean(true) 在 StandardScaler 中的配置是用于决定是否在标准化过程中减去每个特征的均值。这一操作的目的是将数据的均值调整为零，进而进行标准化处理。
    val StandardScaler01=new StandardScaler()
      .setInputCol("price_v")
      .setOutputCol("price_sca")
      .setWithMean(true)

    val StandardScaler02=new StandardScaler()
      .setInputCol("weight_v")
      .setOutputCol("weight_sca")
      .setWithMean(true)

    //  定义管道(按处理数据的)并进行训练
    val pipelineModel:PipelineModel=new Pipeline()
      .setStages(Array(vectorAssembler01,vectorAssembler02,StandardScaler01,StandardScaler02,one_hot_encoder))
      .fit(sku_info)

    //  使用训练的模型进行预测
    val result_data=pipelineModel.transform(sku_info)

    result_data.select("weight_sca").show(false)

    //  one-hot编码之后的会把数据处理为稀疏向量(sparseVector),然后根据题目需要把稀疏向量转化成数组，最后方便展示，只有注册的函数才可以在
    //  sparksql里面直接使用
    spark.udf.register(
      "vectorToArray",
      (r:SparseVector) => {r.toArray.mkString(",")}
    )

    //  标准化处理之后会把数据变成一个密集向量(DenseVector)，然后我们要取到第一个值，才是我们题目要求的数据
    spark.udf.register(
      "vectorToDouble",
      (r: DenseVector) => { r.apply(0) }
    )

    //  将注册的函数代入进去对稀疏向量和密集向量进行转化，最后收集
    val result:Array[Row]=result_data
      .withColumn("spu_id_hot",expr("vectorToArray(spu_id_hot)"))
      .withColumn("tm_id_hot",expr("vectorToArray(tm_id_hot)"))
      .withColumn("category3_id_hot",expr("vectorToArray(category3_id_hot)"))
      .withColumn("price_sca",expr("vectorToDouble(price_sca)"))
      .withColumn("weight_sca",expr("vectorToDouble(weight_sca)"))
      .select("id","price_sca","weight_sca","spu_id_hot","tm_id_hot","category3_id_hot")
      .orderBy("id")
      .limit(1)
      .collect()

    //  最后根据题目的要求打印结果
    println("-------------------第一条数据前10列展示为--------------------")
    result.foreach{
      r => println(r.mkString(",").split(",").take(10).mkString(","))
    }



















    spark.close()
  }

}
